const config = require('config');
const DatabaseManager = require('./DatabaseManager');
const MyDAO = require('./my_dao');
const Consumer = require('./consumer');
const Executor = require('./executor');

// 从配置文件或环境变量中获取Kafka和MongoDB的配置信息
const clientId = process.env.kafka_ClientId || config.kafka.ClientId;
const brokers = (process.env.kafka_brokers ? JSON.parse(process.env.kafka_brokers) : "") || config.kafka.brokers;
const groupId = process.env.kafka_groupId || config.kafka.groupId;
const topic = process.env.kafka_topic || config.kafka.topic || 'Output';
const mongoUrl = process.env.mongo_url || config.mongo.url;
const mongoDb = process.env.mongo_db || config.mongo.db;
const mongoRecordCollection = process.env.mongo_record_collection || config.mongo.record_collection;
const mongoErrorCollection = process.env.mongo_error_collection || config.mongo.error_collection;

// 初始化数据库连接管理器
const dbManager = new DatabaseManager(mongoUrl, mongoDb);
const myDAO = new MyDAO(dbManager, mongoRecordCollection, mongoErrorCollection);
const executor = new Executor();
const consumer = new Consumer(clientId, brokers, groupId, topic, myDAO, executor);
consumer.start().catch(console.error);
